#pragma once

#include <memory>
#include <boost/asio/awaitable.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/redirect_error.hpp>
#include <boost/asio/experimental/channel.hpp>
#include <boost/asio/experimental/concurrent_channel.hpp>

#include "core/stream_session.hpp"
#include "rtmp_handshake.hpp"
#include "rtmp_protocol/rtmp_chunk_protocol.hpp"
#include "server/rtmp/rtmp_message/rtmp_message.hpp"

namespace mms {
    class SocketInterface;
    class RtmpMessage;
    class RtmpConnectCommandMessage;
    class RtmpPublishMessage;
    class RtmpPlayMessage;
    class RtmpMediaSource;
    class RtmpMediaSink;

    class RtmpServerSession : public StreamSession
    {
    public:
        RtmpServerSession(std::shared_ptr<SocketInterface> conn);
        virtual ~RtmpServerSession();

        // session
        void service() override;
        void close() override;

    protected:
        // 同步发送rtmp消息
        template <typename T>
        boost::asio::awaitable<bool> send_rtmp_message(T &msg)
        {
            boost::system::error_code ec;
            std::shared_ptr<RtmpMessage> rtmp_msg = msg.encode();
            if (!rtmp_msg)
            {
                co_return false;
            }
            boost::asio::experimental::channel<void(boost::system::error_code, bool)> wait_channel(conn_->get_worker()->get_io_context());
            std::vector<std::shared_ptr<RtmpMessage>> v = {rtmp_msg};
            co_await rtmp_msgs_channel_.async_send(boost::system::error_code{}, v, &wait_channel, boost::asio::use_awaitable);
            co_await wait_channel.async_receive(boost::asio::redirect_error(
                boost::asio::use_awaitable, ec));
            co_return true;
        }
        // 异步发送
        boost::asio::awaitable<bool> send_rtmp_message(const std::vector<std::shared_ptr<RtmpMessage>> &rtmp_msgs);

    private:
        void start_recv_coroutine();
        boost::asio::awaitable<void> stop_recv_coroutine();

        void start_send_coroutine();
        boost::asio::awaitable<void> stop_send_coroutine();

        boost::asio::awaitable<bool> on_recv_rtmp_message(std::shared_ptr<RtmpMessage> &rtmp_msg);

        boost::asio::awaitable<bool> send_acknowledge_msg_if_required();

        boost::asio::awaitable<bool> handle_amf0_command(std::shared_ptr<RtmpMessage> rtmp_msg);
        boost::asio::awaitable<bool> handle_amf0_connect_command(std::shared_ptr<RtmpMessage> rtmp_msg);

        boost::asio::awaitable<bool> handle_amf0_release_stream_command(std::shared_ptr<RtmpMessage> rtmp_msg);
        boost::asio::awaitable<bool> handle_amf0_fcpublish_command(std::shared_ptr<RtmpMessage> rtmp_msg);
        boost::asio::awaitable<bool> handle_amf0_createstream_command(std::shared_ptr<RtmpMessage> rtmp_msg);
        boost::asio::awaitable<bool> handle_amf0_publish_command(std::shared_ptr<RtmpMessage> rtmp_msg);
         boost::asio::awaitable<bool> handle_amf0_play_command(std::shared_ptr<RtmpMessage> rtmp_msg);
        bool parse_connect_cmd(RtmpConnectCommandMessage &connect_command);
        bool parse_publish_cmd(RtmpPublishMessage &pub_cmd);

        bool parse_play_cmd(RtmpPlayMessage &play_cmd);

        bool handle_video_msg(std::shared_ptr<RtmpMessage> msg);
        bool handle_audio_msg(std::shared_ptr<RtmpMessage> msg);
        bool handle_amf0_data(std::shared_ptr<RtmpMessage> rtmp_msg);

    private:
        void detach_source(); // 将数据迁移到其他session，可能是重推

    private:
        std::shared_ptr<SocketInterface> conn_;
        RtmpHandshake handshkae_;
        RtmpChunkProtocol chunk_protocol_;
        std::atomic_flag closed_ = ATOMIC_FLAG_INIT;
        bool recv_coroutine_running_ = false;
        boost::asio::experimental::concurrent_channel<void(boost::system::error_code, bool)> recv_coroutine_exited_;

        // 发送协程
        bool send_coroutine_running_ = false;
        boost::asio::experimental::concurrent_channel<void(boost::system::error_code, bool)> send_coroutine_exited_;
        using SYNC_CHANNEL = boost::asio::experimental::channel<void(boost::system::error_code, bool)>;
        // SYNC_CHANNEL 用于同步发送
        boost::asio::experimental::channel<void(boost::system::error_code, std::vector<std::shared_ptr<RtmpMessage>>, SYNC_CHANNEL *)> rtmp_msgs_channel_;

        bool is_publisher_ = false;
        std::shared_ptr<RtmpMediaSource> rtmp_media_source_;

        bool is_player_ = false;
        std::shared_ptr<RtmpMediaSink> rtmp_media_sink_;
};


}
